package org.acghub.mtdb.core;

import org.acghub.mtdb.common.Bean;
import org.acghub.mtdb.common.Pair;
import org.acghub.mtdb.common.exception.MtdbException;
import org.acghub.mtdb.common.exception.MtdbKeyVerifyException;

import java.util.Objects;

/**
 * @author zhouhe
 */
public abstract class TTable<K, V> extends AbstractTable<K, V> {

    @Override
    public boolean exists(K key) {
        Objects.requireNonNull(key, "key is null");
        Lockey lockey = Lockeys.INSTANCE.getLockey(tableLockId, (Comparable) key);

        Transaction transaction = Transaction.current();
        transaction.rAddLockey(lockey);

        // 1.transaction local
        TRecord<K, V> cachedTRecord = transaction.getCachedTRecord(meta().getName(), key);
        if (cachedTRecord != null && cachedTRecord.exist()) {
            return true;
        }

        //2.cache
        if (cache.exists(key)) {
            return true;
        }

        //3.storage
        return storage != null && storage.exists(key);
    }

    private V find(K key, boolean readOnly) {
        Objects.requireNonNull(key, "key is null");
        Lockey lockey = Lockeys.INSTANCE.getLockey(tableLockId, (Comparable) key);
        Transaction transaction = Transaction.current();
        if (readOnly) {
            transaction.rAddLockey(lockey);
        } else {
            transaction.wAddLockey(lockey);
        }
        countFind.increment();
        TRecord<K, V> cachedTRecord = transaction.getCachedTRecord(meta().getName(), key);
        if (cachedTRecord != null) {
            return cachedTRecord.getValue();
        }
        Lockey cacheLockey = Lockeys.INSTANCE.getLockey(-tableLockId, (Comparable) key);
        cacheLockey.wLock();
        try {
            TRecord<K, V> r = cache.get(key);
            if (null == r) {
                countFindMiss.increment();
                V value = find0(key);
                if (null == value) {
                    countFindStorageMiss.increment();
                    return null;
                }
                r = new TRecord<>(this, value, lockey, TRecord.State.INDB_GET);
                cache.add(key, r, false);
            }
            transaction.addCachedTRecord(meta().getName(), r);
            return r.getValue();
        } finally {
            cacheLockey.wUnlock();
        }
    }

    @Override
    public V select(K key) {
        return find(key, true);
    }

    @Override
    public V get(K key) {
        return find(key, false);
    }

    @Override
    public boolean insert(K key, V value) {
        Objects.requireNonNull(key, "key is null");
        Objects.requireNonNull(value, "value is null");
        Lockey lockey = Lockeys.INSTANCE.getLockey(tableLockId, (Comparable) key);
        Transaction.current().wAddLockey(lockey);
        //验证key唯一性
        if (!verify(key)) {
            throw new MtdbKeyVerifyException("key verify unique error, key:" + key);
        }
        countAdd.increment();
        TRecord<K, V> r = cache.get(key);
        if (null != r) {
            return r.add(value);
        }
        countAddMiss.increment();
        if (exists0(key)) {
            countAddStorageMiss.increment();
            return false;
        }
        cache.add(key, new TRecord<>(this, value, lockey, TRecord.State.ADD), true);
        return true;
    }

    @Override
    public boolean delete(K key) {
        Objects.requireNonNull(key, "key is null");
        Transaction current = Transaction.current();
        Lockey lockey = Lockeys.INSTANCE.getLockey(tableLockId, (Comparable) key);

        current.wAddLockey(lockey);
        current.removeCachedTRecord(meta().getName(), key);

        countRemove.increment();
        TRecord<K, V> r = cache.get(key);
        if (null != r) {
            return r.remove();
        }
        countRemoveMiss.increment();
        boolean exists = exists0(key);
        if (!exists) {
            countRemoveStorageMiss.increment();
        }
        TRecord<K, V> record = new TRecord<>(this, null, lockey, exists ? TRecord.State.INDB_REMOVE : TRecord.State.REMOVE);
        cache.add(key, record, true);
        return exists;
    }

    @Override
    public void truncate() {
        try {
            Tables.wFlushLock();
        } catch (InterruptedException e){
            Thread.currentThread().interrupt();
            return;
        }
        try {
            cache.truncate();
            storage.truncate();
            Tables.wFlushUnlock();
        } catch (Exception e) {
            Trace.error("truncate error", e);
        } finally {
            Tables.wFlushUnlock();
        }
    }

    @Override
    public long count() {
        if (storage == null) {
            return cache.size();
        } else {
            return storage.count();
        }
    }


    @Override
    public K minKey() {
        org.acghub.mtdb.common.Table table = (org.acghub.mtdb.common.Table) meta.getParent();
        if (storage == null) {
            throw new MtdbException("Table \"" + table.getName() + "\" storage is null.");
        }
        return storage.minKey();
    }

    @Override
    public K maxKey() {
        org.acghub.mtdb.common.Table table = (org.acghub.mtdb.common.Table) meta.getParent();
        if (storage == null) {
            throw new MtdbException("Table \"" + table.getName() + "\" storage is null.");
        }
        return storage.maxKey();
    }

    @Override
    public K nextKey() {
        if (!meta().getKeyType().supportAutoIncrement()) {
            throw new MtdbException("Table key type \"" + meta().getKeyType().getName() + "\" is not support auto increment.");
        }
        org.acghub.mtdb.common.Table table = (org.acghub.mtdb.common.Table) meta.getParent();
        if (storage == null) {
            throw new MtdbException("Table \"" + table.getName() + "\" storage is null.");
        }
        return storage.nextKey();
    }

    @Override
    public V insert(K k) {
        if (!(meta().getValueType() instanceof Bean)) {
            throw new MtdbException("only bean type is supported.");
        }
        V v = newValue();
        if (v == null) {
            throw new MtdbException("newValue is null.");
        }
        if (!insert(k, v)) {
            throw new MtdbException("insert failed.");
        }
        return v;
    }

    @Override
    public Pair<K, V> insert() {
        K k = nextKey();
        if (k == null) {
            throw new MtdbException("key is null.");
        }
        V v = insert(k);
        if (v == null) {
            throw new MtdbException("value is null.");
        }
        return Pair.of(k, v);
    }

    @Override
    public void onChanged(TRecord<K, V> r) {
        if (r.state() == TRecord.State.REMOVE) {
            cache.remove(r.getKey());
        }
        if (storage != null) {
            storage.onChanged(r);
        }
    }

    // for bean change
    final void onChanged(TRecord<K, V> r, LogNotify ln) {
        Transaction.current().addLastCommitActions(() -> onChanged(r));
    }

    // for cache add or remove
    final void onChanged(TRecord<K, V> r, boolean cacheCommit, TRecord.State state) {
        Trace.debug("r={}, cacheCommit={}, state", r, cacheCommit, state);
        Transaction.current().addLastCommitActions(() -> onChanged(r));
    }
}
